package com.winchain.chainbill.bill.biz.transfer.rx;

import com.shell.bee.persistence.transaction.Transaction;
import com.shell.bee.persistence.transaction.TransactionUtil;
import com.shell.framework.rx.OnErrorNotImplementedSubscriber;
import com.winchain.chainbill.bill.base.po.CbXfe;
import com.winchain.chainbill.bill.biz.Bill;
import com.winchain.chainbill.bill.biz.BillTranEvent;
import com.winchain.chainbill.bill.biz.send.rx.BatchBillEvent;
import com.winchain.chainbill.bill.biz.send.vo.BatchBillStats;
import com.winchain.chainbill.bill.biz.transfer.service.BillXfeEventService;
import java.util.List;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import rx.Observable;

@Component
public class BillXfeRevRejectSubscriber extends OnErrorNotImplementedSubscriber<BatchBillEvent<String, CbXfe>> {
	@Autowired
	private BillXfeEventService billXfeEventService;
	@Autowired
	private BillXfeRevRejectPrntSubscriber billXfeRevRejectPrntSubscriber;
	@Autowired
	private BillXfeRevRejectChdSubscriber billXfeRevRejectChdSubscriber;
	@Autowired
	private BillXfeRevRejectMsgSubscriber billXfeRevRejectMsgSubscriber;
	@Autowired
	private BillXfeRevRejectAssetSubscriber billXfeRevRejectAssetSubscriber;

	public void onApplicationEvent(BatchBillEvent<String, CbXfe> t) {
		final String remark = (String) t.getValue("remark");
		List<String> records = t.getRecs();
		BatchBillStats<CbXfe> stats = t.getStats();

		final Bill<CbXfe> source = new Bill();
		for (final String record : records) {
			try {
				TransactionUtil.execute(new Transaction() {
					public void doInTransaction() {
						String billId = record;
						BillTranEvent<CbXfe> event = BillXfeRevRejectSubscriber.this.billXfeEventService
								.getBillTranEvent(billId, false);
						Bill<CbXfe> bill = event.getBill();
						source.setRec(bill.getRec());
						source.setSource(bill.getSource());

						event.setValue("remark", remark);

						Observable<BillTranEvent<CbXfe>> observers = Observable.just(event);
						observers.subscribe(BillXfeRevRejectSubscriber.this.billXfeRevRejectPrntSubscriber);
						observers.subscribe(BillXfeRevRejectSubscriber.this.billXfeRevRejectChdSubscriber);
						observers.subscribe(BillXfeRevRejectSubscriber.this.billXfeRevRejectMsgSubscriber);
						observers.subscribe(BillXfeRevRejectSubscriber.this.billXfeRevRejectAssetSubscriber);
					}
				}, new int[0]);

				stats.addPassedRecord(source);
			} catch (Exception e) {
				e.printStackTrace();
				stats.addAbnormalRecord(source);
				stats.addAbnormalMsg(e.getMessage());
			}
		}
	}
}
